package dacp.etl.kafka.hdfs.connect.utils;

import java.io.File;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Collection;
import java.util.UUID;
import java.util.regex.Matcher;

import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.PathFilter;
import org.apache.kafka.common.TopicPartition;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.google.common.base.Strings;
import com.google.common.collect.Lists;

import dacp.etl.kafka.hdfs.connect.names.FileName;
import io.confluent.connect.hdfs.filter.CommittedFileFilter;
import io.confluent.connect.hdfs.storage.Storage;

public class FileUtils {
	private static final Logger log = LoggerFactory.getLogger(FileUtils.class);

	public static String committedFileName(FileName namer, String url, String topicsDir, String directory,
			TopicPartition topicPart, long startOffset, long endOffset, String extension) {

		String filename = namer.generateName(topicPart, startOffset, endOffset, directory, extension);
		return fileName(url, topicsDir, directory, filename);
	}

	public static String committedFileName(FileName namer, String url, String topicsDir, String directory,
			TopicPartition topicPart, String extension) {
		return committedFileName(namer, url, topicsDir, directory, topicPart, 0L, 0L, extension);
	}

	public static String logFileName(String url, String logsDir, TopicPartition topicPart) {
		return fileName(url, logsDir, topicPart, "log");
	}

	public static String directoryName(String url, String topicsDir, TopicPartition topicPart) {
		String topic = topicPart.topic();
		int partition = topicPart.partition();
		return url + "/" + topicsDir + "/" + topic + "/" + partition;
	}

	public static String fileName(String url, String topicsDir, TopicPartition topicPart, String name) {
		String topic = topicPart.topic();
		int partition = topicPart.partition();
		return url + "/" + topicsDir + "/" + topic + "/" + partition + "/" + name;
	}

	public static String fileName(String url, String topicsDir, String directory, String name) {
		ArrayList<String> piece = Lists.newArrayList();
		if (!Strings.isNullOrEmpty(url)) {
			if(url.endsWith(File.separator)){
				url = url.substring(0, url.length()-1);
			}
			piece.add(url);
		}
		if(!Strings.isNullOrEmpty(topicsDir)){
			if(topicsDir.endsWith(File.separator)){
				topicsDir = topicsDir.substring(0, topicsDir.length()-1);
			}
			piece.add(topicsDir);
		}
		if(!Strings.isNullOrEmpty(directory)){
			if(directory.endsWith(File.separator)){
				directory = directory.substring(0, directory.length()-1);
			} 
			piece.add(directory);
		}
		if(!Strings.isNullOrEmpty(name)){
			piece.add(name);
		}
		return fileName(piece);
	}
	 

	public static String directoryName(String url, String topicsDir, String directory) {
		return url + "/" + topicsDir + "/" + directory;
	}

	public static String tempFileName(String url, String topicsDir, String directory, String extension) {
		UUID id = UUID.randomUUID();
		String name = id.toString() + "_" + "tmp" + extension;
		return fileName(url, topicsDir, directory, name);
	}

	public static String fileName(Collection piece) {
		return StringUtils.join(piece, File.separator);
	}

	public static String topicDirectory(String url, String topicsDir, String topic) {
		return url + "/" + topicsDir + "/" + topic;
	}

	private static ArrayList<FileStatus> traverseImpl(Storage storage, Path path, PathFilter filter)
			throws IOException {
		if (!storage.exists(path.toString())) {
			return new ArrayList<>();
		}
		ArrayList<FileStatus> result = new ArrayList<>();
		FileStatus[] statuses = storage.listStatus(path.toString());
		for (FileStatus status : statuses) {
			if (status.isDirectory()) {
				result.addAll(traverseImpl(storage, status.getPath(), filter));
			} else {
				if (filter.accept(status.getPath())) {
					result.add(status);
				}
			}
		}
		return result;
	}

	public static FileStatus[] traverse(Storage storage, Path path, PathFilter filter) throws IOException {
		ArrayList<FileStatus> result = traverseImpl(storage, path, filter);
		return result.toArray(new FileStatus[result.size()]);
	}

	private static ArrayList<FileStatus> getDirectoriesImpl(Storage storage, Path path) throws IOException {
		FileStatus[] statuses = storage.listStatus(path.toString());
		ArrayList<FileStatus> result = new ArrayList<>();
		for (FileStatus status : statuses) {
			if (status.isDirectory()) {
				int count = 0;
				FileStatus[] fileStatuses = storage.listStatus(status.getPath().toString());
				for (FileStatus fileStatus : fileStatuses) {
					if (fileStatus.isDirectory()) {
						result.addAll(getDirectoriesImpl(storage, fileStatus.getPath()));
					} else {
						count++;
					}
				}
				if (count == fileStatuses.length) {
					result.add(status);
				}
			}
		}
		return result;
	}

	public static FileStatus[] getDirectories(Storage storage, Path path) throws IOException {
		ArrayList<FileStatus> result = getDirectoriesImpl(storage, path);
		return result.toArray(new FileStatus[result.size()]);
	}

	private static ArrayList<FileStatus> traverseImpl(FileSystem fs, Path path) throws IOException {
		if (!fs.exists(path)) {
			return new ArrayList<>();
		}
		ArrayList<FileStatus> result = new ArrayList<>();
		FileStatus[] statuses = fs.listStatus(path);
		for (FileStatus status : statuses) {
			if (status.isDirectory()) {
				result.addAll(traverseImpl(fs, status.getPath()));
			} else {
				result.add(status);
			}
		}
		return result;
	}

	public static FileStatus[] traverse(FileSystem fs, Path path) throws IOException {
		ArrayList<FileStatus> result = traverseImpl(fs, path);
		return result.toArray(new FileStatus[result.size()]);
	}
}
